(ns metabase.warehouse-schema.models.table-test
  (:require
   [clojure.java.jdbc :as jdbc]
   [clojure.test :refer :all]
   [metabase.models.interface :as mi]
   [metabase.models.serialization :as serdes]
   [metabase.permissions-rest.data-permissions.graph :as data-perms.graph]
   [metabase.permissions.core :as perms]
   [metabase.permissions.models.data-permissions :as data-perms]
   [metabase.permissions.models.permissions-group :as perms-group]
   [metabase.sync.core :as sync]
   [metabase.test :as mt]
   [metabase.test.data.one-off-dbs :as one-off-dbs]
   [metabase.test.fixtures :as fixtures]
   [metabase.util :as u]
   [metabase.warehouse-schema.models.field-values :as field-values]
   [metabase.warehouse-schema.models.table :as table]
   [toucan2.core :as t2]))

(use-fixtures :once (fixtures/initialize :db :test-users))

(deftest valid-field-order?-test
  (testing "A valid field ordering is a set IDs  of all active fields in a given table"
    (is (#'table/valid-field-order? (mt/id :venues)
                                    [(mt/id :venues :name)
                                     (mt/id :venues :category_id)
                                     (mt/id :venues :latitude)
                                     (mt/id :venues :longitude)
                                     (mt/id :venues :price)
                                     (mt/id :venues :id)])))
  (testing "Field ordering is invalid if some fields are missing"
    (is (false? (#'table/valid-field-order? (mt/id :venues)
                                            [(mt/id :venues :category_id)
                                             (mt/id :venues :latitude)
                                             (mt/id :venues :longitude)
                                             (mt/id :venues :price)
                                             (mt/id :venues :id)]))))
  (testing "Field ordering is invalid if some fields are from a differnt table"
    (is (false? (#'table/valid-field-order? (mt/id :venues)
                                            [(mt/id :venues :name)
                                             (mt/id :venues :category_id)
                                             (mt/id :venues :latitude)
                                             (mt/id :venues :longitude)
                                             (mt/id :venues :price)
                                             (mt/id :checkins :id)]))))
  (testing "Only active fields should be considerd when checking field order"
    (one-off-dbs/with-blank-db
      (doseq [statement [;; H2 needs that 'guest' user for QP purposes. Set that up
                         "CREATE USER IF NOT EXISTS GUEST PASSWORD 'guest';"
                         ;; Keep DB open until we say otherwise :)
                         "SET DB_CLOSE_DELAY -1;"
                         ;; create table & load data
                         "DROP TABLE IF EXISTS \"BIRDS\";"
                         "CREATE TABLE \"BIRDS\" (\"SPECIES\" VARCHAR PRIMARY KEY, \"EXAMPLE_NAME\" VARCHAR);"
                         "GRANT ALL ON \"BIRDS\" TO GUEST;"
                         (str "INSERT INTO \"BIRDS\" (\"SPECIES\", \"EXAMPLE_NAME\") VALUES "
                              "('House Finch', 'Marshawn Finch'),  "
                              "('California Gull', 'Steven Seagull'), "
                              "('Chicken', 'Colin Fowl');")]]
        (jdbc/execute! one-off-dbs/*conn* [statement]))
      (sync/sync-database! (mt/db))
      (is (#'table/valid-field-order? (mt/id :birds)
                                      [(mt/id :birds :species)
                                       (mt/id :birds :example_name)]))
      (jdbc/execute! one-off-dbs/*conn* ["ALTER TABLE \"BIRDS\" DROP COLUMN \"EXAMPLE_NAME\";"])
      (sync/sync-database! (mt/db))
      (is (#'table/valid-field-order? (mt/id :birds)
                                      [(mt/id :birds :species)])))))

(deftest slashes-in-schema-names-test
  (testing "Schema names should allow forward or back slashes (#8693, #12450)"
    (doseq [schema-name ["my\\schema"
                         "my\\\\schema"
                         "my/schema"
                         "my\\/schema"
                         "my\\\\/schema"]]
      (testing (format "Should be able to create/delete Table with schema name %s" (pr-str schema-name))
        (mt/with-temp [:model/Table {table-id :id} {:schema schema-name}]
          (is (= schema-name
                 (t2/select-one-fn :schema :model/Table :id table-id))))))))

(deftest identity-hash-test
  (testing "Table hashes are composed of the schema name, table name and the database's identity-hash"
    (mt/with-temp [:model/Database db    {:name "field-db" :engine :h2}
                   :model/Table    table {:schema "PUBLIC" :name "widget" :db_id (:id db)}]
      (let [db-hash (serdes/identity-hash db)]
        (is (= "0395fe49"
               (serdes/raw-hash ["PUBLIC" "widget" db-hash])
               (serdes/identity-hash table)))))))

(deftest set-new-table-permissions!-test
  (testing "New permissions are set appropriately for a new table, for all groups"
    (mt/with-full-data-perms-for-all-users!
      (mt/with-temp [:model/PermissionsGroup {group-id :id} {}
                     :model/Database         {db-id :id}    {}
                     :model/Table            {table-id-1 :id} {:db_id  db-id
                                                               :schema "PUBLIC"}
                     :model/Table            {table-id-2 :id} {:db_id  db-id
                                                               :schema "PUBLIC"}]
        ;; Perms for new tables are the same as the DB-level perms, if they exist
        (let [all-users-group-id (u/the-id (perms-group/all-users))]
          (is (partial=
               {all-users-group-id
                {db-id
                 {:perms/view-data             :unrestricted
                  :perms/create-queries        :query-builder-and-native
                  :perms/download-results      :one-million-rows
                  :perms/manage-table-metadata :no
                  :perms/manage-database       :no}}}
               (data-perms.graph/data-permissions-graph :group-id all-users-group-id :db-id db-id))))

        ;; A new group starts with the same perms as All Users
        (is (partial=
             {group-id
              {db-id
               {:perms/view-data             :unrestricted
                :perms/create-queries        :query-builder-and-native
                :perms/download-results      :one-million-rows
                :perms/manage-table-metadata :no
                :perms/manage-database       :no}}}
             (data-perms.graph/data-permissions-graph :group-id group-id :db-id db-id)))

        (testing "A new table has appropriate defaults, when perms are already set granularly for the DB"
          (data-perms/set-table-permission! group-id table-id-1 :perms/create-queries :no)
          (data-perms/set-table-permission! group-id table-id-1 :perms/download-results :no)
          (data-perms/set-table-permission! group-id table-id-1 :perms/manage-table-metadata :no)
          (data-perms/set-table-permission! group-id table-id-2 :perms/create-queries :query-builder)
          (data-perms/set-table-permission! group-id table-id-2 :perms/download-results :one-million-rows)
          (data-perms/set-table-permission! group-id table-id-2 :perms/manage-table-metadata :yes)
          (mt/with-temp [:model/Table {table-id-3 :id} {:db_id  db-id
                                                        :schema "PUBLIC"}]
            (is (partial=
                 {group-id
                  {db-id
                   {:perms/view-data             :unrestricted
                    :perms/create-queries        {"PUBLIC"
                                                  {table-id-1 :no
                                                   table-id-2 :query-builder
                                                   table-id-3 :no}}
                    :perms/download-results      {"PUBLIC"
                                                  {table-id-1 :no
                                                   table-id-2 :one-million-rows
                                                   table-id-3 :no}}
                    :perms/manage-table-metadata {"PUBLIC"
                                                  {table-id-1 :no
                                                   table-id-2 :yes
                                                   table-id-3 :no}}
                    :perms/manage-database       :no}}}
                 (data-perms.graph/data-permissions-graph :group-id group-id :db-id db-id)))))))))

(deftest cleanup-permissions-after-delete-table-test
  (mt/with-temp
    [:model/Database {db-id :id}      {}
     :model/Table    {table-id-1 :id} {:db_id db-id}
     :model/Table    {}               {:db_id db-id}]
    (data-perms/set-table-permission! (perms-group/all-users) table-id-1 :perms/create-queries :query-builder-and-native)
    (data-perms/set-table-permission! (perms-group/all-users) table-id-1 :perms/view-data :unrestricted)
    (data-perms/set-table-permission! (perms-group/all-users) table-id-1 :perms/download-results :one-million-rows)
    (data-perms/set-table-permission! (perms-group/all-users) table-id-1 :perms/manage-table-metadata :yes)
    (is (true? (t2/exists? :model/DataPermissions :table_id table-id-1)))
    (t2/delete! :model/Table table-id-1)
    (testing "Table-level permissions are deleted when we delete the table"
      (is (false? (t2/exists? :model/DataPermissions :table_id table-id-1))))))

;; hydration tests
(deftest field-values-hydration-test
  ;; Manually activate Field values since they are not created during sync (#53387)
  (field-values/get-or-create-full-field-values! (t2/select-one :model/Field :id (mt/id :venues :price)))
  (field-values/get-or-create-full-field-values! (t2/select-one :model/Field :id (mt/id :venues :name)))

  (is (=? {(mt/id :venues :price) (mt/malli=? [:sequential {:min 1} :any])
           (mt/id :venues :name)  (mt/malli=? [:sequential {:min 1} :any])}
          (-> (t2/select-one :model/Table (mt/id :venues))
              (t2/hydrate :field_values)
              :field_values))))

(deftest pk-field-hydration-test
  (is (= (mt/id :venues :id)
         (-> (t2/select-one :model/Table (mt/id :venues))
             (t2/hydrate :pk_field)
             :pk_field))))

(deftest visible-filter-clause-table-level-test
  (testing "visible-tables-filter-clause generates a HoneySQL clause that filters tables based on user permissions"
    (mt/with-no-data-perms-for-all-users!
      (mt/with-temp [:model/Database {db-id :id} {}
                     :model/Table    {table-view-data-unrestricted-create-queries-query-builder-id :id} {:db_id db-id :name "Table1"}
                     :model/Table    {table-view-data-unrestricted-create-queries-query-builder-and-native-id :id} {:db_id db-id :name "Table2"}
                     :model/Table    {table-view-data-blocked-create-queries-no-id :id} {:db_id db-id :name "Table3"}
                     :model/Table    {table-view-data-legacy-no-self-service-create-queries-no-id :id} {:db_id db-id :name "Table4"}
                     :model/PermissionsGroup pg1 {}
                     :model/PermissionsGroup pg2 {}]

        (perms/add-user-to-group! (mt/user->id :rasta) pg1)
        (perms/add-user-to-group! (mt/user->id :rasta) pg2)

        (t2/delete! :model/DataPermissions :db_id db-id)

        (data-perms/set-database-permission! pg1 db-id :perms/view-data :blocked)
        (data-perms/set-database-permission! pg1 db-id :perms/create-queries :no)

        (data-perms/set-database-permission! pg2 db-id :perms/view-data :legacy-no-self-service)
        (data-perms/set-database-permission! pg2 db-id :perms/create-queries :no)

        (data-perms/set-table-permission! pg1 table-view-data-unrestricted-create-queries-query-builder-id :perms/view-data :unrestricted)
        (data-perms/set-table-permission! pg1 table-view-data-unrestricted-create-queries-query-builder-id :perms/create-queries :query-builder)

        (data-perms/set-table-permission! pg1 table-view-data-unrestricted-create-queries-query-builder-and-native-id :perms/view-data :unrestricted)
        (data-perms/set-table-permission! pg1 table-view-data-unrestricted-create-queries-query-builder-and-native-id :perms/create-queries :query-builder-and-native)

        (data-perms/set-table-permission! pg1 table-view-data-blocked-create-queries-no-id :perms/view-data :blocked)
        (data-perms/set-table-permission! pg1 table-view-data-blocked-create-queries-no-id :perms/create-queries :no)

        (data-perms/set-table-permission! pg1 table-view-data-legacy-no-self-service-create-queries-no-id :perms/view-data :legacy-no-self-service)
        (data-perms/set-table-permission! pg1 table-view-data-legacy-no-self-service-create-queries-no-id :perms/create-queries :no)

        (let [user-id            (mt/user->id :rasta)
              fetch-visible-ids (fn [user-info permission-mapping table-id-field]
                                  (let [filter-clause (mi/visible-filter-clause :model/Table table-id-field user-info permission-mapping)]
                                    (t2/select-pks-set [:model/Table]
                                                       {:where
                                                        [:and
                                                         [:= :db_id db-id]
                                                         filter-clause]})))]

          (testing "Superuser should see all tables"
            (let [user-info        {:user-id       user-id ; User ID doesn't matter for superuser
                                    :is-superuser? true}
                  permission-map {:perms/view-data      :unrestricted ; Levels don't matter for superuser
                                  :perms/create-queries :query-builder}]
              (is (= #{table-view-data-unrestricted-create-queries-query-builder-id
                       table-view-data-unrestricted-create-queries-query-builder-and-native-id
                       table-view-data-blocked-create-queries-no-id
                       table-view-data-legacy-no-self-service-create-queries-no-id}
                     (fetch-visible-ids user-info permission-map :id)))))

          (testing "Non-superuser should only see tables where they have both view and query permissions (mimics mi/can-read?)"
            (let [user-info      {:user-id       user-id
                                  :is-superuser? false}
                  permission-map {:perms/view-data      :unrestricted
                                  :perms/create-queries :query-builder}]
              (is (= #{table-view-data-unrestricted-create-queries-query-builder-id
                       table-view-data-unrestricted-create-queries-query-builder-and-native-id}
                     (fetch-visible-ids user-info permission-map :id))
                  "Clause should filter out tables without sufficient view/query permissions")

              (testing "using a sequence of fields for the table ID"
                (is (= #{table-view-data-unrestricted-create-queries-query-builder-id
                         table-view-data-unrestricted-create-queries-query-builder-and-native-id}
                       (fetch-visible-ids user-info permission-map [:coalesce :id :metabase_table.id]))
                    "Clause should work with coalesce when a sequence of fields is provided"))

              (testing "using a qualified keyword"
                (is (= #{table-view-data-unrestricted-create-queries-query-builder-id
                         table-view-data-unrestricted-create-queries-query-builder-and-native-id}
                       (fetch-visible-ids user-info permission-map :metabase_table.id))
                    "Clause should work with qualified keyword"))))

          (testing "Non-superuser requiring only :view-data :unrestricted should see tables 1 and 2"
            (let [user-info           {:user-id       user-id
                                       :is-superuser? false}
                  permission-map      {:perms/view-data      :unrestricted
                                       :perms/create-queries :no}]
              (is (= #{table-view-data-unrestricted-create-queries-query-builder-id
                       table-view-data-unrestricted-create-queries-query-builder-and-native-id}
                     (fetch-visible-ids user-info permission-map :id))
                  "Clause should respect the provided permission levels")))

          (testing "Non-superuser requiring only :view-data :legacy-no-self-service should see tables 1, 2, and 4"
            (let [user-info           {:user-id       user-id
                                       :is-superuser? false}
                  permission-map      {:perms/view-data      :legacy-no-self-service
                                       :perms/create-queries :no}]
              (is (= #{table-view-data-unrestricted-create-queries-query-builder-id
                       table-view-data-unrestricted-create-queries-query-builder-and-native-id
                       table-view-data-legacy-no-self-service-create-queries-no-id}
                     (fetch-visible-ids user-info permission-map :id))
                  "Clause should respect the provided permission levels")))

          (testing "Non-superuser requiring :view-data :blocked should see all tables since all values are more permissive "
            (let [user-info           {:user-id       user-id
                                       :is-superuser? false}
                  permission-map      {:perms/view-data      :blocked
                                       :perms/create-queries :no}]
              (is (= #{table-view-data-unrestricted-create-queries-query-builder-id
                       table-view-data-unrestricted-create-queries-query-builder-and-native-id
                       table-view-data-blocked-create-queries-no-id
                       table-view-data-legacy-no-self-service-create-queries-no-id}
                     (fetch-visible-ids user-info permission-map :id))
                  "Clause should filter correctly when requiring :blocked level")))

          (testing "Blocked table takes precedence over legacy-no-self-service when in any group a user is a member of"
            (data-perms/set-table-permission! pg2 table-view-data-legacy-no-self-service-create-queries-no-id :perms/view-data :blocked)
            (data-perms/set-table-permission! pg2 table-view-data-legacy-no-self-service-create-queries-no-id :perms/create-queries :no)

            (let [user-info           {:user-id       user-id
                                       :is-superuser? false}
                  permission-map      {:perms/view-data      :legacy-no-self-service
                                       :perms/create-queries :no}]
              (is (= #{table-view-data-unrestricted-create-queries-query-builder-id
                       table-view-data-unrestricted-create-queries-query-builder-and-native-id}
                     (fetch-visible-ids user-info permission-map :id))
                  "Clause should respect the provided permission levels"))))))))

(deftest visible-filter-clause-db-level-test
  (testing "visible-tables-filter-clause generates a HoneySQL clause that filters tables based on user permissions when only db level perms are set"
    (mt/with-temp [:model/Database {db-id :id} {}
                   :model/Table    {table-id-1 :id} {:db_id db-id :name "Table1"}
                   :model/Table    {table-id-2 :id} {:db_id db-id :name "Table2"}
                   :model/Table    {table-id-3 :id} {:db_id db-id :name "Table3"}]

      (data-perms/set-database-permission! (perms-group/all-users) db-id :perms/view-data :unrestricted)
      (data-perms/set-database-permission! (perms-group/all-users) db-id :perms/create-queries :no)

      (let [user-id            (mt/user->id :rasta)
            fetch-visible-ids (fn [user-cfg permission-map table-id-field]
                                (let [filter-clause (mi/visible-filter-clause :model/Table table-id-field user-cfg permission-map)]
                                  (t2/select-pks-set [:model/Table]
                                                     {:where
                                                      [:and
                                                       [:= :db_id db-id]
                                                       filter-clause]})))]

        (testing "Superuser should see all tables"
          (let [superuser-cfg  {:user-id       user-id ; User ID doesn't matter for superuser
                                :is-superuser? true}
                permission-map {:perms/view-data       :unrestricted ; Levels don't matter for superuser
                                :perms/create-queries :query-builder}]
            (is (= #{table-id-1 table-id-2 table-id-3}
                   (fetch-visible-ids superuser-cfg permission-map :id)))))

        (testing "Non-superuser should only see tables where they have both view and query permissions (mimics mi/can-read?)"
          (let [user-info      {:user-id       user-id
                                :is-superuser? false}
                permission-map {:perms/view-data      :unrestricted
                                :perms/create-queries :query-builder}]
            (is (empty?                  ; Expecting empty set if no tables are visible with these perms
                 (fetch-visible-ids user-info permission-map :id))
                "Clause should filter out tables without sufficient view/query permissions")

            (testing "using an expression for the table ID"
              (is (empty?
                   (fetch-visible-ids user-info permission-map [:coalesce :id :metabase_table.id]))
                  "Clause should work with coalesce when a sequence of fields is provided"))

            (testing "using a qualified keyword"
              (is (empty?
                   (fetch-visible-ids user-info permission-map :metabase_table.id))
                  "Clause should work with qualified keyword"))))

        (testing "Non-superuser requiring only :view-data :unrestricted should see tables 1, 2 and 3"
          (let [user-info      {:user-id       user-id
                                :is-superuser? false}
                permission-map {:perms/view-data      :unrestricted
                                :perms/create-queries :no}]
            (is (= #{table-id-1 table-id-2 table-id-3}
                   (fetch-visible-ids user-info permission-map :id))
                "Clause should respect the provided permission levels")))

        (testing "Non-superuser requiring :view-data :blocked should all tables since all values are more permissive "
          (let [user-info      {:user-id       user-id
                                :is-superuser? false}
                permission-map {:perms/view-data      :blocked
                                :perms/create-queries :no}]
            (is (= #{table-id-1 table-id-2 table-id-3}
                   (fetch-visible-ids user-info permission-map :id))
                "Clause should filter correctly when requiring :blocked level")))))))

(deftest prevent-metabase-transform-data-source-change-test
  (testing "Cannot change data_source from metabase-transform"
    (mt/with-temp [:model/Table {table-id :id} {:data_source :metabase-transform}]
      (testing "to another value"
        (is (thrown-with-msg?
             clojure.lang.ExceptionInfo
             #"Cannot change data_source from metabase-transform"
             (t2/update! :model/Table table-id {:data_source :transform}))))
      (testing "to nil"
        (is (thrown-with-msg?
             clojure.lang.ExceptionInfo
             #"Cannot change data_source from metabase-transform"
             (t2/update! :model/Table table-id {:data_source nil}))))))

  (testing "Cannot change data_source to metabase-transform"
    (mt/with-temp [:model/Table {table-id :id} {:data_source :ingested}]
      (testing "from another value"
        (is (thrown-with-msg?
             clojure.lang.ExceptionInfo
             #"Cannot set data_source to metabase-transform"
             (t2/update! :model/Table table-id {:data_source :metabase-transform}))))
      (testing "but can change to other non-metabase-transform values"
        (is (some? (t2/update! :model/Table table-id {:data_source :ingested})))
        (is (= :ingested (t2/select-one-fn :data_source :model/Table :id table-id))))
      (testing "can also change it to nil"
        (is (some? (t2/update! :model/Table table-id {:data_source nil})))
        (is (nil? (t2/select-one-fn :data_source :model/Table :id table-id)))))))

(deftest is-published-and-collection-id-test
  (testing "is_published defaults to false"
    (mt/with-temp [:model/Table {table-id :id} {}]
      (is (false? (t2/select-one-fn :is_published :model/Table :id table-id)))))
  (testing "can create a table with is_published=true and collection_id"
    (mt/with-temp [:model/Collection {coll-id :id} {:name "Test Collection"}
                   :model/Table {table-id :id} {:is_published true :collection_id coll-id}]
      (let [table (t2/select-one :model/Table :id table-id)]
        (is (true? (:is_published table)))
        (is (= coll-id (:collection_id table))))))
  (testing "collection_id FK constraint prevents referencing non-existent collection"
    (is (thrown?
         Exception
         (mt/with-temp [:model/Table _ {:collection_id Integer/MAX_VALUE}]))))
  (testing "deleting a collection unpublishes the tables in it"
    (mt/with-temp [:model/Collection {coll-id :id} {:name "Test Collection"}
                   :model/Table {table-1-id :id} {:is_published true :collection_id coll-id}
                   :model/Table {table-2-id :id} {:is_published true :collection_id coll-id}]
      (t2/delete! :model/Collection :id coll-id)
      (is (= #{[false nil]} (t2/select-fn-set (juxt :is_published :collection_id) :model/Table
                                              :id [:in [table-1-id table-2-id]]))))))

(deftest collection-hydration-test
  (testing "hydrating :collection on a table"
    (mt/with-temp [:model/Collection {coll-id :id} {:name "Test Collection"}
                   :model/Table table {:is_published true :collection_id coll-id}]
      (let [hydrated (t2/hydrate table :collection)]
        (is (= coll-id (-> hydrated :collection :id)))
        (is (= "Test Collection" (-> hydrated :collection :name))))))
  (testing "hydrating :collection on a table with no collection_id returns nil"
    (mt/with-temp [:model/Table table {}]
      (let [hydrated (t2/hydrate table :collection)]
        (is (nil? (:collection hydrated))))))
  (testing "batched hydration works for multiple tables"
    (mt/with-temp [:model/Collection {coll1-id :id} {:name "Collection 1"}
                   :model/Collection {coll2-id :id} {:name "Collection 2"}
                   :model/Table table1 {:is_published true :collection_id coll1-id}
                   :model/Table table2 {:is_published true :collection_id coll2-id}
                   :model/Table table3 {}]
      (let [hydrated (t2/hydrate [table1 table2 table3] :collection)]
        (is (= coll1-id (-> hydrated first :collection :id)))
        (is (= coll2-id (-> hydrated second :collection :id)))
        (is (nil? (-> hydrated (nth 2) :collection)))))))
